package com.example.mq.server;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.server.core.*;
import com.example.mq.server.dao.DiskDataCenter;
import com.example.mq.server.dao.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class VirtualHost {
    private String virtualHostName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);
    // 锁对象
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    public VirtualHost(String name) {
        this.virtualHostName = name;
        diskDataCenter.init();
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (Exception e) {
            System.out.println("[VirtualHost] 恢复内存失败!");
            e.printStackTrace();
        }
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    // 交换机添加操作
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        // 交换机添加操作
        // 1.将交换机与虚拟主机产生联系与其他虚拟主机隔离
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 2.从内存中查询是否含有交换机,有就不创建了,没有在创建
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    // 说明已经创建过了
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName = " + exchangeName);
                    return true;
                }
                // 3.创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setAutoDelete(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 4.判断是否需要持久化
                if (durable) {
                    // 写入硬盘
                    diskDataCenter.insertExchange(exchange);
                }
                // 5.结果写入内存
                memoryDataCenter.addExchange(exchange);
            }
            System.out.println("[VirtualHost] 交换机添加完成! exchangeName = " + exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机添加失败! exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 交换机删除操作
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if (exchange == null) {
                    throw new MqException("[virtualHost] 交换机不存在无法删除!");
                }
                // 判断时候是持久化的,持久化的还需要删除硬盘数据
                if (exchange.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
            }
            System.out.println("[VirtualHost] 交换机删除成功! exchangeName = " + exchangeName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 队列添加操作
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        // 拼接队列名字
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                // 1.判断是否已经存在
                if (existsQueue != null) {
                    // 有就不创建了
                    System.out.println("[VirtualHost] 队列已经存在! queueName = " + queueName);
                    return true;
                }
                // 没有则创建
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 2.判断是否要持久化
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                // 3.将结果写入内存
                memoryDataCenter.addQueue(queue);
            }
            System.out.println("[VirtualHost] 队列添加成功! queueName = " + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列添加失败! queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 队列删除操作
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            synchronized (queueLocker) {
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在无法删除!");
                }
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
            }
            System.out.println("[VirtualHost] 队列删除成功! queueName = " + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueBind(String exchangeName, String queueName, String bindingKey) {
        // 拼接名字
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1.查询绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] 绑定已经存在! queueName = " + queueName +
                                ",exchangeName = " + exchangeName);
                    }
                    // 2.判断bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey非法! bindingKey = " + bindingKey);
                    }
                    // 3.创建binding对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 4.查询队列交换机是否存在!
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在!");
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在!");
                    }
                    // 5.添加绑定
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.addBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName = " + exchangeName +
                    ",QueueName = " + queueName);
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName = " + exchangeName +
                    ",QueueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueUnbind(String exchangeName, String queueName) {
        // 拼接名字
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        // 绑定不存在
                        throw new MqException("[VirtualHost] 绑定不存在！exchangeName = " + exchangeName +
                                ",QueueName = " + queueName);
                    }
                    // 直接删除绑定
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 删除绑定成功!exchangeName = " + exchangeName +
                    ",QueueName = " + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败!exchangeName = " + exchangeName +
                    ",QueueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 1.拼接交换机的名字
            exchangeName = virtualHostName + exchangeName;
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在！exchangeName = " + exchangeName);
            }
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey非法！");
            }
            // 2.查看交换机的类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                // direct直接转发
                String queueName = virtualHostName + routingKey;
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！queueName = " + queueName);
                }
                Message message = Message.createMessageWithId(basicProperties,routingKey,body);
                sendMessage(queue,message);
            } else {
                // fanout、topic转发
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 获取绑定对象，判断队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        System.out.println("[VirtualHost] basicPublish 队列不存在");
                        // 有可能是队列已经删除，但是绑定没有删除，跳过这个绑定关系即可
                        continue;
                    }
                    Message message = Message.createMessageWithId(basicProperties, routingKey, body);
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发送失败！");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode = message.getDeliverMode();
        if (deliverMode == 2) {
            // 消息进行持久化
            diskDataCenter.sendMessage(queue, message);
        }
        // 消息写入内存
        memoryDataCenter.sendMessage(queue, message);
        // 通知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    // 队列的订阅者
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        queueName = virtualHostName + queueName;
        try {
            // 对应的队列添加消费者
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume成功！queueName = " + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume失败！queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 手动ack
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            // 1.获取队列和消息
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 待确认的队列不存在！queueName = " + queueName);
            }
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 待确认的消息不存在！messageId = " + messageId);
            }
            // 2.删除硬盘消息
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 3.删除消息中心的消息
            memoryDataCenter.deleteMessage(messageId);
            // 4.删除待确认的消息
            memoryDataCenter.deleteMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck成功！消息确认成功！queueName = " + queueName +
                    ",messageId = " + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck失败！消息确认失败！queueName = " + queueName +
                    ",messageId = " + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
